use std::time::SystemTime;
use std::time::UNIX_EPOCH;

use anyhow::anyhow;
use anyhow::Result;
use clap::Args;
use fudata::db;
use fudata::db::FUDB;
use fudata::model::trade::trade_order::TradeOrder;
use log::debug;
use log::error;
use log::info;
use surrealdb::engine::remote::ws;
use surrealdb::method::Stream;
use futures::StreamExt;
use zenoh::prelude::r#async::*;
use zenoh::queryable::Queryable;

use crate::cli::cnf::zenohconf::CommonArgs;
use crate::cli::commands::register_ctrl_c;
use crate::cli::commands::wait_until_next_scheduled_time;
use crate::qtrade::model::heartbeat::Heartbeat;
use crate::qtrade::model::heartbeat::Server;
use crate::qtrade::worker_query;
use crate::qtrade::MAX_WORKERS;
use crate::qtrade::TRADE_HOLDINGS_TOPIC;
use crate::qtrade::TRADE_ORDERS_TOPIC;
use crate::qtrade::TRADE_STRATEGIES_TOPIC;
use crate::qtrade::ZENOH_HEARTBEAT_TOPIC;
use tokio::select;

#[derive(Args, Debug)]
pub struct QtrCommandArguments {
    #[arg(
        help = "Database path used for storing data",
        short = 'p',
        long = "path",
        requires = "path",
        default_value = "memory"
    )]
    pub path: String,

    #[arg(
        help = "settings to craw securities",
        short = 's',
        long = "settings",
        default_value = "./config.toml"
    )]
    /// settings in toml format, e.g. config.toml
    pub settings: String,
    #[command(flatten)]
    common: CommonArgs,
}

pub async fn init(QtrCommandArguments {
    path,
    settings,
    common,
}: QtrCommandArguments) -> Result<()> {
    let dbrs = db::init().await;
    match dbrs {
        Ok(_) => {}
        Err(e) => {
            return Err(anyhow!(e));
        }
    }

    info!("Starting fudacraw qtr...");

    debug!("(qtr::init) arguments: path= {}", path);
    debug!("(qtr::init) arguments: settings=  {}", settings);

    info!("Opening session...");

    let _config = zenoh::config::Config::default();
    let config: zenoh::config::Config = common.into();
    let session: std::sync::Arc<Session> = zenoh::open(config).res().await.unwrap().into_arc();
    let session_clone1 = session.clone();

    // Listen for Ctrl-C signal
    let shutdown_rx_arc = register_ctrl_c();
    let query_injector = worker_query::start_worker_pool(MAX_WORKERS);

    let mut main_state = MainState::new(&session).await?;
    let mut heartbeat_count = 0;
    loop {
        // Wrap the receiver in a cloneable Arc<Mutex<_>> to allow concurrent access
        let mut shutdown_rx_locked = shutdown_rx_arc.lock().await;

        select!(
            _ = shutdown_rx_locked.recv()  => {
                info!("Shutting down from signal handler");
                break;
            }

            // server heartbeat, publish heartbeat every 5 seconds to remote peers
            heartbeat_time = wait_until_next_scheduled_time("0/5 * * * * *","heartbeat_time" )=>{
                let server_hb = Heartbeat::Server(Server {
                    id: format!("svid{}", heartbeat_count),
                    svid: format!("svid{}", heartbeat_count),
                    timestamp_ms: get_timestamp_ms(),
                    desc: "server_hb@qtr.rs".to_string(),
                });
                let json = server_hb.serialize().unwrap();
                info!("\n heartbeat scheduled time: {:?} payload: {:?}", heartbeat_time, json );

                let rs = main_state.heartbeat_publisher.put(json).res().await;
                if rs.is_err() {
                    error!("heartbeat failed: {:?}", rs.err());
                }
                heartbeat_count += 1;
            }

            // read local database: publish live tradecommand updates to remote peers
            live_update_trade_order = main_state.live_trade_order_stream.next() => {
                if let Some(update) = live_update_trade_order {
                    match update {
                        Ok(notification) => {
                            info!(">> (live_update_trade_order) New Data: {:?}", notification);
                            let trade_order = notification.data.clone().to_owned();
                            let payload = trade_order.serialize().unwrap();
                            let rs = main_state.trade_orders_publisher.put(payload).res().await;
                            if rs.is_err(){
                                error!("(live_update_trade_order) put error: {:?}", rs.err());
                            } 
                          
                            // test/test> select * from trade_order limit 1
                            // test/test> update trade_order:01J5D8VS1FTK6A7FDD52G1X45S set kl_volume=266
                            // Further processing or publishing of the data
                        },
                        Err(error) => {
                            error!("Error receiving live TradeOrder update: {:?}", error);
                        },
                    }
                }
            },


            // read local database: provide query queryable_trade_strategies service to remote peers
            queryable_trade_strategies = main_state.queryable_trade_strategies.recv_async() => {
                if let Ok(query) = queryable_trade_strategies {
                    let mut is_other_peer = true; 
                    if let Some(zid) = query.value(){if zid.to_string() == session.zid().to_string(){ is_other_peer=false }};

                    if is_other_peer{
                        // get a query request from remote peer
                        let _=   query_injector.push(query);
                    }
                }
            },

            // read local database: provide query queryable_trade_orders service to remote peers
            queryable_trade_orders = main_state.queryable_trade_orders.recv_async() => {
                if let Ok(query) = queryable_trade_orders {
                    let mut is_other_peer = true; 
                    if let Some(zid) = query.value(){if zid.to_string() == session.zid().to_string(){ is_other_peer=false }};

                    if is_other_peer{
                        // get a query request from remote peer
                        let _=   query_injector.push(query);
                    }
                }
            },

            // read local database: provide query queryable_trade_holdings service to remote peers
            queryable_trade_holdings = main_state.queryable_trade_holdings.recv_async() => {
                if let Ok(query) = queryable_trade_holdings {
                    let mut is_other_peer = true; 
                    if let Some(zid) = query.value(){if zid.to_string() == session.zid().to_string(){ is_other_peer=false }};

                    if is_other_peer{
                        // get a query request from remote peer
                        let _=   query_injector.push(query);
                    }
                }
            },

        );
    }

    // All ok
    Ok(())
}

/// Represents the state needed for managing trade info synchronization.
pub struct MainState<'a> {
    pub heartbeat_publisher: zenoh::publication::Publisher<'a>,
    pub trade_orders_publisher: zenoh::publication::Publisher<'a>,
    pub live_trade_order_stream: Stream<'a, ws::Client, Vec<TradeOrder>>,
    pub queryable_trade_strategies: Queryable<'a, flume::Receiver<zenoh::queryable::Query>>,
    pub queryable_trade_orders: Queryable<'a, flume::Receiver<zenoh::queryable::Query>>,
    pub queryable_trade_holdings: Queryable<'a, flume::Receiver<zenoh::queryable::Query>>,
}

impl<'a> MainState<'a> {
    pub async fn new(session: &'a Session) -> Result<Self> {
        let complete = false;

        let heartbeat_publisher: zenoh::publication::Publisher = session
            .declare_publisher(ZENOH_HEARTBEAT_TOPIC)
            .res().await
            .unwrap();

        let trade_orders_publisher: zenoh::publication::Publisher = session
            .declare_publisher(TRADE_ORDERS_TOPIC)
            .res().await
            .unwrap();

        let live_trade_order_stream: Stream<ws::Client, Vec<TradeOrder>> = FUDB.select(
            "trade_order"
        ).live().await?;

        let queryable_trade_strategies = session
            .declare_queryable(TRADE_STRATEGIES_TOPIC)
            .complete(complete)
            .res().await
            .unwrap();

        let queryable_trade_orders = session
            .declare_queryable(TRADE_ORDERS_TOPIC)
            .complete(complete)
            .res().await
            .unwrap();

        let queryable_trade_holdings = session
            .declare_queryable(TRADE_HOLDINGS_TOPIC)
            .complete(complete)
            .res().await
            .unwrap();

        Ok(Self {
            heartbeat_publisher,
            trade_orders_publisher,
            live_trade_order_stream,
            queryable_trade_strategies,
            queryable_trade_orders,
            queryable_trade_holdings,
        })
    }
}
fn get_timestamp_ms() -> u64 {
    let start = SystemTime::now();
    let since_the_epoch = start.duration_since(UNIX_EPOCH).expect("Time went backwards");
    since_the_epoch.as_millis() as u64
}
